package com.lagou.rpc.consumer.tool;//package com.lagou.rpc.provider2.tool;

import com.lagou.rpc.consumer.proxy.RpcClientProxy;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ZkCli {

    public static ZkClient zkClient = null;
    private static String P_NODE = "/base";
    private static Map<String, String> serverCountMap = new ConcurrentHashMap<>();

    /**
     * 定时任务线程池
     */
    private static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);

    public ZkCli() {
        Connect();
    }

    public static ZkClient getZkClient() {
        if (zkClient == null) {
            Connect();
        }
        return zkClient;
    }

    public static void Connect() {
        zkClient = new ZkClient("127.0.0.1:2181");
        if (!zkClient.exists(P_NODE)) {
            zkClient.createPersistent(P_NODE);
        }
        zkClient.subscribeChildChanges(P_NODE, new IZkChildListener() {
            @Override
            public void handleChildChange(String s, List<String> list) throws Exception {
                System.out.println("s = " + s);
                System.out.println("list = " + list.toString());
                Map<String, Object> tempMap = new ConcurrentHashMap<>();
                for (String node : list) {
                    Object readData = zkClient.readData(P_NODE + "/" + node);
                    tempMap.put(node, readData);
                    if (serverCountMap.get(node) == null) {
                        serverCountMap.put(node, readData.toString());
                        RpcClientProxy.getOneRpcClient(node, readData.toString());
                    }
                }

                serverCountMap.forEach((key, value) -> {
                    if (tempMap.get(key) == null) {
                        serverCountMap.remove(key);
                        RpcClientProxy.removeRpcClient(key);
                    }
                });
            }
        });

        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                List<String> children = zkClient.getChildren(P_NODE);
                long current = System.currentTimeMillis();

                children.forEach((node) -> {
                    String value = zkClient.readData(P_NODE + "/" + node).toString();
                    String[] arr = value.split("#");

                    //ip#port#time#dealTime
                    String ip = arr[0];
                    String port = arr[1];
                    if (arr.length > 2) {
                        long preTime = Long.parseLong(arr[2]);
                        if (current - preTime > 5000) {
                            zkClient.writeData(P_NODE + "/" + ip + "#" + port, ip + "#" + port);
                            System.out.println("服务器：" + ip + "#" + port + "，5秒内没有任何请求的进入，自动清除节点中的时间，清除后节点的值为：" + ip + "#" + port);
                        }
                    }
                });
            }
        }, 5, 5, TimeUnit.SECONDS);
    }

    public static void startConnect() {
        if (zkClient == null) {
            Connect();
        }
        List<String> children = zkClient.getChildren(P_NODE);
        if (children != null && children.size() > 0) {
            children.forEach((node) -> {
                Object readData = zkClient.readData(P_NODE + "/" + node);
                serverCountMap.put(node, readData.toString());
            });
        }
        RpcClientProxy.initRpcClient(serverCountMap);
    }

    public static void setNodeValue(String key, long startTime, long endTime) {
        double time = (endTime - startTime) / 1000;
        String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        zkClient.writeData(P_NODE + "/" + key, key + "#" + endTime + "#" + time + "#" + date);
        System.out.println("请求服务器：" + key + "，响应时间为：" + time + "秒" + ",系统时间：" + date);
        System.out.println("对应节点：" + key + "，的值为：" + zkClient.readData(P_NODE + "/" + key));
    }

    //获取最短响应时间的服务器
    public static String getMinimumResponseTime() {
        String key = null;
        Map<String, String> map = new HashMap<>();
        double responseTime = 1000;
        int counter = 0;
        List<String> children = zkClient.getChildren(P_NODE);
        if (children != null && children.size() > 1) {
            for (String child : children) {
                Object readData = zkClient.readData(P_NODE + "/" + child);
                String[] split = readData.toString().split("#");
                if (split.length > 2) {
                    counter++;
                    double aLong = Double.valueOf(split[3]);
                    if (aLong < responseTime) {
                        responseTime = aLong;
                        key = child;
                    }else if (aLong == responseTime) {
                        key = null;
                    }
                }
            }
        }
        if (counter < 2) {
            key = null;
        }
        return key;
    }
}
